package plgcore

import (
	"dgo/MyProj/PrjConf"
	"dgo/MyProj/PrjHttpUtils"
	"dgo/goutils/utils"
	"dgo/plg/plgnet"
	"fmt"
	"gitee.com/ymofen/supervalue"
)

/*
*

	分发数据时，不进行异步队列投递，直接进行分发, 回调可能会多线程执行
*/
type PlgCoreSyncSubscriber struct {
	ex     *utils.SubscribeCenter
	plugId string
}

func (this *PlgCoreSyncSubscriber) SetPlugID(val string) {
	this.plugId = val
}

func (this *PlgCoreSyncSubscriber) ReloadConf(conf *wrapper.SuperValue) error {

	if this.ex == nil {
		this.ex = utils.NewSubscribeCenter()
		this.ex.DispMode = 2 // 直接分发
		utils.Monitor.AddMonitor(this.ex.Monitor_Exec)
		utils.Monitor.AddMonitor(this.ex.Monitor_Push)
	}

	if conf == nil {
		return nil
	}

	this.ex.AllTopicDispFlag = int8(conf.IntByName("alltopic-disp", 0))

	return nil
}

func (this *PlgCoreSyncSubscriber) GetSubscriber() *utils.SubscribeCenter {
	return this.ex
}

func (this *PlgCoreSyncSubscriber) DispatchData(channelid string, sender interface{}, data interface{}, tag interface{}) (err error) {
	this.ex.DispatchData(channelid, sender, data, tag)
	return nil
}

func (this *PlgCoreSyncSubscriber) OnObjectNotify(sender interface{}, notifytype int, data interface{}) (rval interface{}, err error) {
	if notifytype == utils.NOTIFY_TYPE_SETCONF {
		if conf, ok := data.(*wrapper.SuperValue); ok {
			err = this.ReloadConf(conf)
		} else {
			err = fmt.Errorf("非法的JSON配置")
		}
	} else if notifytype == utils.NOTIFY_TYPE_RELEASE {
		utils.Monitor.RemoveMonitorRec(this.ex.Monitor_Exec)
		utils.Monitor.RemoveMonitorRec(this.ex.Monitor_Push)
	}
	return
}

/*
**

	响应HttpStatus请求
*/
func (this *PlgCoreSyncSubscriber) OnHttpStatus(rec *plgnet.PlgHttpRec) (rval interface{}, errcode int, err error) {
	itype := rec.QryAsInt("type", 0)
	if itype == 1 {
		id := rec.QryAsString("id", "")
		str := this.ex.ChannelStatusString(id)
		str += fmt.Sprintf("%s status\r\n%s", id, str)
		rec.RespW.Write([]byte(str))
	} else if itype == 3 {
		str := this.ex.StatusString()
		rec.RespW.Write([]byte(str))
	} else if itype == 4 {
		str := this.ex.ChannelDetails(20)
		rec.RespW.Write([]byte(str))
	} else if itype == 99 {
		json := wrapper.NewSVObject()
		json.SetKeyValues("plugid", this.plugId, "uuid", PrjConf.APP_UUID,
			"alltopic-disp", this.ex.AllTopicDispFlag,
			"channel", this.ex.ChannelCount(),
			"push", this.ex.Monitor_Push.Info(), "pop", this.ex.Monitor_Exec.Info())
		json.SetKeyValue("desc", "分发数据时，不进行异步队列投递，直接进行分发")
		json.SetKeyValue("now", utils.NowString())
		PrjHttpUtils.JSON2TableRec(rec.RespW, "", json.AsJSONString(false))
	} else {
		var sb utils.BytesBuilder
		PrjHttpUtils.AppendIndexBody(&sb, "status", "status?type=99")
		PrjHttpUtils.AppendIndexBody(&sb, "ex-task-status", "status?type=3")
		PrjHttpUtils.AppendIndexBody(&sb, "ex-channel-details", "status?type=4")
		PrjHttpUtils.AppendIndexBody(&sb, "ex-channel-status", "status?type=1&id=xxx")
		PrjHttpUtils.ResponseBodyHtml(rec.RespW, sb.String())
	}
	return
}

func CreatePlgCoreSyncSubscriberFunc(args ...interface{}) (rval interface{}, err error) {
	obj := &PlgCoreSyncSubscriber{}
	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.plugId = str
		}
	}

	rval = obj
	return
}
